package com.dxwan.observables.creating;


import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.Observer;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.BooleanSupplier;
import org.reactivestreams.Publisher;
import org.reactivestreams.Subscriber;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;

public class App {
    public static void main(String[] args) throws InterruptedException {

        //1. Observable.create
        Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> emitter) throws Exception {
                emitter.onNext("hello");
                emitter.onNext("world");

                emitter.onComplete();
            }
        }).subscribe(new Observer<String>() {
            @Override
            public void onSubscribe(Disposable d) {
            }

            @Override
            public void onNext(String s) {
                System.out.println("now:" + s);
            }

            @Override
            public void onError(Throwable e) {
            }

            @Override
            public void onComplete() {
                System.out.println("c'est fini");
            }
        });

        //2. Observable.defer
        //延迟产生Observable，直到subscribe时才生成
        Observable.defer( ()->{
            Thread.sleep(2000);
            return Observable.just("un","deux","trois","quatr","cinq");
        })
        .subscribe( System.out::println);

        //3. Observable.empty
        //产生一个空的Observable
        Observable.empty()
                .subscribe( System.out::println);

        Observable.never()
                .subscribe( System.out::println);

        Observable.error( new ArithmeticException())
                .subscribe(new Observer<Object>() {
                    @Override
                    public void onSubscribe(Disposable d) {

                    }

                    @Override
                    public void onNext(Object o) {

                    }

                    @Override
                    public void onError(Throwable e) {
                        System.out.println("Error occurred.");
                    }

                    @Override
                    public void onComplete() {

                    }
                });

        //from
        List<String> stringList = new ArrayList<>();
        stringList.add("one");
        stringList.add("two");
        stringList.add("three");
        Observable.fromIterable(stringList)
                .subscribe( System.out::println);


        Observable.fromCallable( ()-> "callable").subscribe( System.out::println);

        Observable.fromArray(new String[]{"yi", "er", "san"})
                .subscribe( System.out::println);

        Observable.fromPublisher(new Publisher<String>() {
            @Override
            public void subscribe(Subscriber<? super String> s) {
                s.onNext("一");
                s.onNext("二");
                s.onNext("三");
            }
        }).subscribe( System.out::println);

        Observable.fromFuture(new Future<String>() {
            @Override
            public boolean cancel(boolean mayInterruptIfRunning) {
                return false;
            }

            @Override
            public boolean isCancelled() {
                return false;
            }

            @Override
            public boolean isDone() {
                return false;
            }

            @Override
            public String get() throws InterruptedException, ExecutionException {
                return "future";
            }

            @Override
            public String get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
                return null;
            }
        })
        .subscribe( System.out::println);

        final CountDownLatch latch = new CountDownLatch(1);
        Observable.interval(0, 100, TimeUnit.MILLISECONDS)
                .subscribe(new Observer<Long>() {
                    @Override
                    public void onSubscribe(Disposable d) {
                    }

                    @Override
                    public void onNext(Long value) {
                        System.out.println(value);

                        if( value == 4)
                            this.onComplete();
                    }

                    @Override
                    public void onError(Throwable e) {
                    }

                    @Override
                    public void onComplete() {
                        latch.countDown();
                    }
                });

        latch.await();

        Observable.just("1.1","1.2","1.3","1.4","1.5")
                .subscribe( System.out::println);

        Observable.range(100,5)
                .subscribe( System.out::println);

        Observable.just(1,2)
                .repeat(3)
                .subscribe( System.out::println);

        Observable.just(1,2)
                .repeatUntil(new BooleanSupplier() {
                    private int times = 0;
                    @Override
                    public boolean getAsBoolean() throws Exception {
                        if( times++ == 5) {
                            return true;
                        }
                        return false;
                    }
                })
                .subscribe( System.out::println);

        CountDownLatch latch2 = new CountDownLatch(1);
        Observable.timer(1, TimeUnit.SECONDS)
                .subscribe(new Observer<Long>() {
                    @Override
                    public void onSubscribe(Disposable d) {

                    }

                    @Override
                    public void onNext(Long aLong) {
                        System.out.println(aLong);

                    }

                    @Override
                    public void onError(Throwable e) {

                    }

                    @Override
                    public void onComplete() {
                        latch2.countDown();

                    }
                });

        latch2.await();

    }
}
